package org.huangrui.spark.scala.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * nc -lp 9999
 * nc -L -p 9999
 *
 * @Author hr
 * @Create 2024-10-21 19:31 
 */
object SparkStreaming05_State {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "huangrui")
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val sc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))
    // 无状态数据操作，只对当前的采集周期内的数据进行处理
    // 在某些场合下，需要保留数据统计结果（状态），实现数据的汇总
    // 使用有状态操作时，需要设定检查点路径，将中间结果保存到文件系统中
    sc.checkpoint("cp")
    val lines = sc.socketTextStream("localhost", 9999)
    val words: DStream[String] = lines.flatMap(_.split(" "))
    val wordOne = words.map(word => (word, 1))
    //    val ds: DStream[(String, Int)] = wordOne.reduceByKey(_ + _)
    // updateStateByKey：根据key对数据的状态进行更新
    // 传递的参数中含有两个值
    // 第一个值表示相同的key的value数据
    // 第二个值表示缓存区相同key的value数据
    val state = wordOne.updateStateByKey(
      (values: Seq[Int], state: Option[Int]) => {
        val newCnt = state.getOrElse(0) + values.sum
        Option(newCnt)
      }
    )

    state.print()

    sc.start()
    sc.awaitTermination()

  }
}
